package com.winit.schedule.task.tracker.remoting;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.event.EventHandler;
import com.winit.schedule.core.protocol.command.HeartbeatResponse;
import com.winit.schedule.core.service.AbstractService;
import com.winit.schedule.remoting.CommandBody;
import com.winit.schedule.remoting.InvokeCallback;
import com.winit.schedule.remoting.exception.RemotingCommandFieldCheckException;
import com.winit.schedule.remoting.netty.NettyClientConfig;
import com.winit.schedule.remoting.netty.NettyRemotingClient;
import com.winit.schedule.remoting.netty.NettyRequestProcessor;
import com.winit.schedule.remoting.protocol.RemotingCommand;
import com.winit.schedule.task.tracker.TaskTrackerContext;
import com.winit.schedule.task.tracker.conf.TaskConfiguration;
import com.winit.schedule.task.tracker.event.ConfigChangeEvent;
import com.winit.schedule.task.tracker.event.ConfigChangeEventType;

public class RemotingClientService extends AbstractService implements EventHandler<ConfigChangeEvent> {

    private static final Logger log     = LoggerFactory.getLogger(RemotingClientService.class);

    private NettyRemotingClient remotingClient;

    private long                invokeTimeoutMillis;

    boolean                     isStart = false;

    private transient String    jobTrackerAddress;

    private TaskTrackerContext  taskTrackerContext;

    public RemotingClientService(TaskTrackerContext taskTrackerContext){
        super("RemotingClientService");
        this.taskTrackerContext = taskTrackerContext;
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {
        super.serviceInit(conf);

        this.remotingClient = new HandlerTimeoutResponseRemotingClient(getNettyClientConfig());

        String jobTrackerAddress = this.getConfig().getString(TaskConfiguration.JT_ADDRESS,
            TaskConfiguration.DEFAULT_JT_ADDRESS);

        log.info("JobTracker addressis is : " + jobTrackerAddress);

        invokeTimeoutMillis = this.getConfig().getLong(TaskConfiguration.CONNECT_TIMEOUT,
            TaskConfiguration.DEFAULT_CONNECT_TIMEOUT);

        log.info("connect timeout is : " + invokeTimeoutMillis + "ms");
    }

    @Override
    protected void serviceStart() throws Exception {
        super.serviceStart();
        remotingClient.start();

        remotingClient.registerDefaultProcessor(new RemotingDispatcher(this), Executors.newCachedThreadPool());
        isStart = true;
    }

    @Override
    protected void serviceStop() throws Exception {
        isStart = false;
        super.serviceStop();
        remotingClient.shutdown();
    }

    /**
     * 同步调用
     * 
     * @param request
     * @return
     * @throws RemotingCommandFieldCheckException
     * @throws JobTrackerNotFoundException
     */
    public RemotingCommand invokeSync(RemotingCommand request) throws RemotingCommandFieldCheckException {

        try {
            request.checkCommandBody();

            RemotingCommand response = remotingClient.invokeSync(getJobTrackerAddress(), request, invokeTimeoutMillis);
            return response;

        } catch (RemotingCommandFieldCheckException e) {
            log.error(e.getMessage(), e);
            throw e;
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
            throw new RemotingCommandFieldCheckException("call RemotingClientService.invokeAsync error.", e);
        }
    }

    /**
     * 异步调用
     * 
     * @param request
     * @param invokeCallback
     * @throws RemotingCommandFieldCheckException
     * @throws JobTrackerNotFoundException
     */
    public void invokeAsync(RemotingCommand request, InvokeCallback invokeCallback)
                                                                                   throws RemotingCommandFieldCheckException {
        try {
            request.checkCommandBody();

            remotingClient.invokeAsync(getJobTrackerAddress(), request, invokeTimeoutMillis, invokeCallback);

        } catch (RemotingCommandFieldCheckException e) {
            log.error(e.getMessage(), e);
            throw e;
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
            throw new RemotingCommandFieldCheckException("call RemotingClientService.invokeAsync error.", e);
        }
    }

    /**
     * 单向调用
     * 
     * @param request
     * @throws RemotingCommandFieldCheckException
     * @throws JobTrackerNotFoundException
     */
    public void invokeOneway(RemotingCommand request) throws RemotingCommandFieldCheckException {
        try {
            request.checkCommandBody();

            remotingClient.invokeOneway(getJobTrackerAddress(), request, invokeTimeoutMillis);

        } catch (RemotingCommandFieldCheckException e) {
            log.error(e.getMessage(), e);
            throw e;
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
            throw new RemotingCommandFieldCheckException("call RemotingClientService.invokeOneway error.", e);
        }
    }

    public void registerProcessor(int requestCode, NettyRequestProcessor processor, ExecutorService executor) {
        remotingClient.registerProcessor(requestCode, processor, executor);
    }

    public void registerDefaultProcessor(NettyRequestProcessor processor, ExecutorService executor) {
        remotingClient.registerDefaultProcessor(processor, executor);
    }

    public NettyRemotingClient getNettyClient() {
        return remotingClient;
    }

    private NettyClientConfig getNettyClientConfig() {
        NettyClientConfig config = new NettyClientConfig();
        return config;
    }

    public boolean isStart() {
        return isStart;
    }

    public long getInvokeTimeoutMillis() {
        return invokeTimeoutMillis;
    }

    public String getJobTrackerAddress() {
        return jobTrackerAddress;
        // return this.getConfig().getString(TaskConfiguration.JT_ADDRESS,
        // TaskConfiguration.DEFAULT_JT_ADDRESS);
    }

    public TaskTrackerContext getTaskTrackerContext() {
        return taskTrackerContext;
    }

    @Override
    public void handle(ConfigChangeEvent event) {
        if (ConfigChangeEventType.MASTER_CHANGE.equals(event.getType())) {
            jobTrackerAddress = event.getMasterAddress();
        }
    }

    class HandlerTimeoutResponseRemotingClient extends NettyRemotingClient {

        public HandlerTimeoutResponseRemotingClient(NettyClientConfig nettyClientConfig){
            super(nettyClientConfig);
        }

        /**
         * 任务拉取响应超时后，能获取到任务项则继续执行拉取的定时任务
         */
        @Override
        public void handlerTimeoutResponse(CommandBody body) {
            if (null != body && body instanceof HeartbeatResponse) {
                HeartbeatResponse response = (HeartbeatResponse) body;
                if (null != response.getTasks() && response.getTasks().size() > 0) {
                    log.info(">>>>>>>The heartbeat response timed out, start the timer task.");
                    
                    getTaskTrackerContext().getTaskTracker().reAllocationTask(response.getTasks());
                }
            }
        }

    }

}
